/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package aiaudio.processing.ratings.inverse;

import aiaudio.processing.base.mapreduce.UserValuePair;
import aiaudio.lastfm.hbase.CannotCreateTableException;
import aiaudio.Application;
import aiaudio.processing.MainTableGroup;
import java.io.IOException;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.mapreduce.Job;

/**
 *
 * @author nastya
 */
public class InverseListenCountAlgorithm {

    private final MainTableGroup mainTableGroup;

    public InverseListenCountAlgorithm(MainTableGroup mainTableGroup) {
        this.mainTableGroup = mainTableGroup;
    }

    public void start() throws IOException, CannotCreateTableException, InterruptedException, ClassNotFoundException {
        Job job = new Job();

        Scan scan = new Scan();

        initMapProcess(scan, job);
        initReducer(job);

        job.waitForCompletion(true);

    }

    private void initReducer(Job job) throws IOException {
        String inversedRatingMatrix = mainTableGroup.getPrepareRatingsAlgTableGroup().getInversedRatingMatrix();
        String inversedRatingsTableName = Application.database().getByName(inversedRatingMatrix).getDatabaseName();
        TableMapReduceUtil.initTableReducerJob(inversedRatingsTableName, RatingInversionReducer.class, job);
    }

    private void initMapProcess(Scan scan, Job job) throws IOException {
        String ratingListTable = mainTableGroup.getInitialDataTableGroup().getRatingListTable();
        String ratingsTableName = Application.database().getByName(ratingListTable).getDatabaseName();
        TableMapReduceUtil.initTableMapperJob(ratingsTableName, scan, ByUserMapper.class, ImmutableBytesWritable.class,
                UserValuePair.class, job);
    }
}
